#!/usr/bin/env python


"""
Unit test for main.py

"""

__revision__ = "$Id$"

import unittest
from test import test_support
import sys
sys.path.insert(0,'../')
import getpass
from StringIO import StringIO
from time import sleep

from psycopg import ProgrammingError
import psycopg

import main

DBNAME = 'test'
USER = getpass.getuser()
INPUT_ENCODING = 'utf-8'

XML = '<?xml version="1.0" encoding="utf-8"?>'

test = (
    '<a/>' ,
    '<a> </a>' ,
    '<a b="c"/>' ,
    '<a b="c" d="e"/>' ,
    '<a><a/></a>' ,
    '<a><a> </a></a>' ,
    '<a><a b="c"/></a>' ,
    '<a><a b="c" d="e"/></a>' ,
    '<a>abc</a>' ,
    '<a><a>abc</a></a>' ,
    '<a b="10"/>' ,
    '<a b="20" d="e"/>' ,
    '<a><a b="30"/></a>' ,
    '<a><a b="40" d="e"/></a>' ,
    '<c><a b="30"/></c>' ,
    '<c><a b="40" d="e"/></c>' ,
    )

class nonclosableStringIO ( StringIO ) :
    def close ( self ) : pass

def reset_db () :
    conn = psycopg.connect ( 'dbname=template1' )
    conn.autocommit(1)
    c = 0
    while not main.autorollback (conn,'drop database '+DBNAME ) :
        c += 1
        sleep(.1)
        if c == 10 : break
    main.autocommit (conn,'create database '+DBNAME )
    conn.close()

def exec_main ( argv , stdin=None , stdout=None , stderr=None ) :
    sysargv = sys.argv
    sys.argv = argv
    sysstdin,sysstdout,sysstderr = sys.stdin,sys.stdout,sys.stderr
    if stdin is not None : sys.stdin = stdin
    if stdout is not None : sys.stdout = stdout
    if stderr is not None : sys.stderr = stderr
    main.main(INPUT_ENCODING,USER)
    sys.argv = sysargv
    sys.stdin,sys.stdout,sys.stderr = sysstdin,sysstdout,sysstderr

class main_base (unittest.TestCase) :
    def setUp ( self ) : reset_db()
        
class Test_main_help (main_base) :
    def test_main ( self ) :
        argv = ( 'test' , '-h' )
        stdout = StringIO()
        self.assertRaises(SystemExit,exec_main,argv,None,stdout)
        self.assert_(stdout.getvalue().startswith(
            'usage: test [OPTIONS]... [[DBNAME] USER] [FILE]..'))

class Test_main_info (main_base) :
    def test_main ( self ) :
        argv = [ 'test' ,  '--init' , '--quiet' , DBNAME ]
        exec_main ( argv )
        argv = ( 'test' ,  '--info' , DBNAME )
        stdout = nonclosableStringIO()
        stdin = StringIO('<?xml version="1.0" encoding="utf-8"?><a/>')
        stderr = StringIO()
        exec_main(argv,stdin,stdout,stderr)
        self.assertEqual(stdout.getvalue()[:31],
                         "Available tables:\nUsed tables:\n")
    
class Test_main_list (main_base) :      # XXX
    def test_main ( self ) :
        argv = ( 'test' ,  '--list' , DBNAME )
        self.assertRaises(NotImplementedError,exec_main,argv)
    
class Test_main_pack_names (main_base) :
    argv = [( 'test' ,  '--pack=names' , DBNAME )]
    def setUp ( self ) :
        super(Test_main_pack_names,self).setUp()
        exec_main ( ('test' ,  '--init' , '--quiet' , DBNAME) )
        for x in test : exec_main(('test','--quiet',DBNAME,USER),
                                  StringIO(XML+x))
        for av in self.argv : exec_main(av)
    def test_main ( self ) :
        for i,x in enumerate(test) :
            stdout = nonclosableStringIO()
            exec_main(('test','--extract',str(i+1),DBNAME,USER),None,stdout)
            self.assertEqual(XML+x,stdout.getvalue())

class Test_main_pack_values (Test_main_pack_names) :
    argv = [( 'test' ,  '--pack=names' , DBNAME ) ,
            ( 'test' ,  '--pack=values' , DBNAME )]
    
class Test_main_pack_pairs (Test_main_pack_names) :
    argv = [( 'test' ,  '--pack=names' , DBNAME ) ,
            ( 'test' ,  '--pack=values' , DBNAME ),
            ( 'test' ,  '--pack=pairs' , DBNAME )]
    
class Test_main_pack_lists (Test_main_pack_names) :
    argv = [( 'test' ,  '--pack=names' , DBNAME ) ,
            ( 'test' ,  '--pack=values' , DBNAME ),
            ( 'test' ,  '--pack=pairs' , DBNAME ),
            ( 'test' ,  '--pack=lists' , DBNAME )]

class Test_main_fill (main_base) :
    def setUp ( self ) :
        super(Test_main_fill,self).setUp()
        exec_main(('test','--init',DBNAME,USER))
        self.conn = psycopg.connect ( 'dbname=%s user=%s' % (DBNAME,USER) )
        self.cr = self.conn.cursor()
        self.cr.execute(
            'create table table_name ( attr_list bigint )' )
        self.conn.commit()
        self.conn.close()
    def test_main ( self ) :
        argv = ( 'test' ,  '--fill=table_name' , DBNAME )
        self.assert_(exec_main(argv) is None)
    
class Test_main_free (Test_main_fill) :
    def test_main ( self ) :
        argv = ( 'test' ,  '--free=table_name' , DBNAME )
        exec_main(argv)

    
class Test_main_init (main_base) :
    def test_main ( self ) :
        argv = ( 'test' ,  '--init' , '--quiet' , DBNAME )
        exec_main(argv)
    
class Test_main_init_schema (main_base) :
    def setUp ( self ) :
        super(Test_main_init_schema,self).setUp()
        argv = ( 'test' , '--schema' , 'tmp' , '--init' , '--quiet' , DBNAME )
        exec_main(argv)
        self.conn = psycopg.connect ( 'dbname=%s user=%s' % (DBNAME,USER) )
    def tearDown ( self ) :
        self.conn.rollback()
        self.conn.close()
        super(Test_main_init_schema,self).tearDown()
    def test_main ( self ) :
        cr = self.conn.cursor()
        cr.execute ("""
        select count(*) from information_schema.tables
        where table_schema = 'tmp' and table_name='fragment_forest'
        """ )
        self.assertEqual(cr.fetchone()[0],1)
        
class Test_main_extract (main_base) :
    def setUp ( self ) :
        super(Test_main_extract,self).setUp()
        argv = [ 'test' ,  '--init' , '--quiet' , DBNAME ]
        exec_main(argv)
        stdin = StringIO('<?xml version="1.0" encoding="utf-8"?><a/>')
        argv = ( 'test' , '--quiet' , DBNAME )
        exec_main ( argv , stdin )
    def test_main ( self ) :
        argv = ( 'test' ,  '--extract' , '1' , DBNAME )
        stdout = nonclosableStringIO()
        exec_main ( argv , stdout=stdout )
        self.assertEqual(stdout.getvalue(),
                         '<?xml version="1.0" encoding="utf-8"?><a/>')
    
class Test_main_insert (main_base) :
    def setUp ( self ) :
        super(Test_main_insert,self).setUp()
        argv = [ 'test' ,  '--init' , '--quiet' , DBNAME ]
        exec_main(argv)
    def test_main ( self ) :
        stdin = StringIO('<?xml version="1.0" encoding="utf-8"?><a/>')
        stderr = StringIO()
        argv = ( 'test' , DBNAME )
        exec_main ( argv , stdin , None , stderr )
        self.assertEqual(stderr.getvalue(),"Standard input :\t1\n")

def test_main():
    suite = unittest.TestSuite()
    suite.addTest(unittest.makeSuite(Test_main_help))
    suite.addTest(unittest.makeSuite(Test_main_info))
    suite.addTest(unittest.makeSuite(Test_main_list))
    suite.addTest(unittest.makeSuite(Test_main_pack_names))
    suite.addTest(unittest.makeSuite(Test_main_pack_values))
    suite.addTest(unittest.makeSuite(Test_main_pack_pairs))
    #suite.addTest(unittest.makeSuite(Test_main_pack_lists))
    suite.addTest(unittest.makeSuite(Test_main_fill))
    suite.addTest(unittest.makeSuite(Test_main_free))
    suite.addTest(unittest.makeSuite(Test_main_init))
    suite.addTest(unittest.makeSuite(Test_main_init_schema))
    suite.addTest(unittest.makeSuite(Test_main_insert))
    suite.addTest(unittest.makeSuite(Test_main_extract))
    test_support.run_suite(suite)

if __name__ == "__main__" :
    test_main()
#===========================================================
